package com.atguigu.flink.chapter05.Source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * TODO 读文件
 *
 * @author cjp
 * @version 1.0
 * @date 2021/8/7 16:31
 */
public class Flink03_Kafka {
    public static void main(String[] args) throws Exception {
        // 0. 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop1:9092,hadoop2:9092,hadoop3:9092");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "aaa210323a");

        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                "flink210323",
                new SimpleStringSchema(),
                properties);

//        kafkaConsumer
//                .setStartFromEarliest();
////                .setStartFromLatest();
////                .setStartFromGroupOffsets();

        DataStreamSource<String> kafkaDS = env.addSource(kafkaConsumer);

        kafkaDS.print();


        env.execute();


    }
}
/*
    回顾： Kafka的offset重置策略
        earliest： 从上次的offset接着消费，如果offset不存在，那么就从头消费
        latest：   从上次的offset接着消费，如果offset不存在，只消费 消费者启动后 产生的数据


    setStartFromEarliest()          =>  不管有没有 offset，一定从头消费
    setStartFromLatest()            =>  不管有没有offset， 一定从 启动后的最新数据 消费
    setStartFromGroupOffsets()      =>  优先从指定的 offset接着消费，如果没有offset，那么从 最新的！ 消费
        =》 这个是默认的策略
    setStartFromTimestamp           =>  从指定的时间点接着消费
    setStartFromSpecificOffsets     =>  从指定的 offset 接着消费
 */